//
// Created by hjie on 23-6-24.
//

#include "publish_subscribe_module.h"
#include "./../protocol/msg_key.h"
#include "message_router.h"
#include "logic_room_data.h"
#include "./../data/data_define.h"

PublishSubscribeModule::PublishSubscribeModule(const std::string &module_name)
{
    m_module_name = module_name;
}

void PublishSubscribeModule::RegisterMsg(MessageRouter *router)
{
    REGISTER_MSG_TYPE(publish_req_k, PublishSubscribeModule, PublishReq, this);
    REGISTER_MSG_TYPE(publish_stream_req_k, PublishSubscribeModule, PublishStream, this);
    REGISTER_MSG_TYPE(get_publish_stream_status_req_k, PublishSubscribeModule, GetPublishStreamStatus, this);
    REGISTER_MSG_TYPE(subscribe_req_k, PublishSubscribeModule, SubscribeReq, this);
    REGISTER_MSG_TYPE(send_answer_req_k, PublishSubscribeModule, SendAnswerReq, this);
    REGISTER_MSG_TYPE(ice_candidate_req_k, PublishSubscribeModule, IceCandidateReq, this);

    m_msg_type_set.insert(publish_req_k);
    m_msg_type_set.insert(subscribe_req_k);

    m_msg_type_set.insert(publish_stream_req_k);
    m_msg_type_set.insert(get_publish_stream_status_req_k);

    m_msg_type_set.insert(send_answer_req_k);

    m_msg_type_set.insert(ice_candidate_req_k);
}

std::shared_ptr<BaseProtocol> PublishSubscribeModule::CreateProtocolData(const std::string &msg, rapidjson::Document &input)
{
    if (msg == publish_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new PublishReqProtocol());
    }
    else if (msg == publish_stream_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new PublishStreamReqProtocol());
    }
    else if (msg == get_publish_stream_status_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new GetPublishStreamStatusReqProtocol());
    }
    else if (msg == subscribe_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new SubscribeReqProtocol());
    }
    else if (msg == send_answer_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new SendAnswerReqProtocol());
    }
    else if (msg == ice_candidate_req_k)
    {
        return std::shared_ptr<BaseProtocol>(new IceCandidateReqProtocol());
    }
    return nullptr;
}

void PublishSubscribeModule::PublishReq(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    auto * req_protocol = (PublishReqProtocol*)request.get();

    std::string & target_user_id = req_protocol->m_user_id;
    std::string & sdp_type       = req_protocol->m_sdp_type;
    std::string & sdp_info       = req_protocol->m_sdp_info;

    std::string & user_id        = data->m_connection_data->m_user_name;

    auto publish_user   = data->m_room->m_data_manager->m_ua_mgr->FindUserAssociation(user_id);
    auto subscribe_user = data->m_room->m_data_manager->m_ua_mgr->FindUserAssociation(target_user_id);

    bool result = false;
    auto response = new PublishResProtocol();
    if (publish_user && publish_user->m_is_publish_stream)
    {
        auto iter = publish_user->m_publish_set.find(target_user_id);
        if (iter == publish_user->m_publish_set.end())
        {
            result = false;
        }
        else
        {
            result = true;
        }
    }
    if (result && subscribe_user)
    {
        auto iter = subscribe_user->m_subscribe_set.find(user_id);
        if (iter == subscribe_user->m_subscribe_set.end())
        {
            response->MakeResponseHeaderProtocol(publish_res_k, data->m_room->RoomId(), RESPONSE, -1);
        }
        else
        {
            response->MakeResponseHeaderProtocol(publish_res_k, data->m_room->RoomId(), RESPONSE, 0);

            auto target = new PublishTargetProtocol();
            target->MakeResponseHeaderProtocol(publish_target_k, data->m_room->RoomId(), TARGET, 0);
            target->m_response_header->m_from_user_id = user_id;
            target->m_response_header->m_to_user_id   = target_user_id;

            target->m_from_user_id = data->m_connection_data->m_user_name;
            target->m_sdp_info     = sdp_info;
            target->m_sdp_type     = sdp_type;

            data->m_response_vec.PushBack(target);
        }
    }
    else
    {
        response->MakeResponseHeaderProtocol(publish_res_k, data->m_room->RoomId(), RESPONSE, -1);
    }
    data->m_response_vec.PushBack(response);
}

void PublishSubscribeModule::SubscribeReq(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    auto * protocol = (SubscribeReqProtocol*)request.get();

    std::string & target_user_id = protocol->m_subscribe_user_id;
    std::string & owner_user_id  = protocol->m_request_header->m_user_id;

    auto response = new SubscribeResProtocol();
    response->MakeResponseHeaderProtocol(subscribe_res_k, data->m_room->RoomId(), RESPONSE, 0);

    data->m_response_vec.PushBack(response);

    auto target = new SubscribeTargetProtocol();
    target->MakeResponseHeaderProtocol(subscribe_target_k, data->m_room->RoomId(), TARGET, 0);
    target->m_response_header->m_to_user_id   = target_user_id;
    target->m_response_header->m_from_user_id = owner_user_id;
    target->m_subscribe_user_id = owner_user_id;

    data->m_room->m_data_manager->m_ua_mgr->AddPublishTarget(target_user_id, owner_user_id);
    data->m_room->m_data_manager->m_ua_mgr->AddSubscribe(owner_user_id, target_user_id);

    data->m_response_vec.PushBack(target);
}

void PublishSubscribeModule::PublishStream(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    auto protocol = (PublishStreamReqProtocol*)request.get();

    std::string user_id = data->m_connection_data->m_user_name;
    data->m_room->m_data_manager->m_ua_mgr->PublishStream(user_id, protocol->m_status);

    auto response = new PublishStreamResProtocol();
    response->MakeResponseHeaderProtocol(publish_stream_res_k, data->m_room->RoomId(), RESPONSE, 0);

    data->m_response_vec.PushBack(response);
}

void PublishSubscribeModule::GetPublishStreamStatus(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    printf("%s\n\n", __func__ );
    auto protocol = (GetPublishStreamStatusReqProtocol*)request.get();
    std::string & search_user_id = protocol->m_user_id;

    auto user_association = data->m_room->m_data_manager->m_ua_mgr->FindUserAssociation(search_user_id);

    auto response = new GetPublishStreamStatusResProtocol();
    response->MakeResponseHeaderProtocol(get_publish_stream_status_res_k, data->m_room->RoomId(), RESPONSE, 0);
    if (user_association)
    {
        response->m_user_id = search_user_id;
        response->m_publish_status = user_association->m_is_publish_stream;
    }
    else
    {
        response->m_user_id = search_user_id;
        response->m_publish_status = false;
    }
    data->m_response_vec.PushBack(response);
}

void PublishSubscribeModule::SendAnswerReq(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    auto * req_protocol = (SendAnswerReqProtocol*)request.get();

    std::string & target_user_id = req_protocol->m_user_id;
    std::string & sdp_type       = req_protocol->m_type;
    std::string & sdp_info       = req_protocol->m_sdp;

    std::string & user_id        = data->m_connection_data->m_user_name;

    auto subscribe_user = data->m_room->m_data_manager->m_ua_mgr->FindUserAssociation(user_id);
    auto publish_user   = data->m_room->m_data_manager->m_ua_mgr->FindUserAssociation(target_user_id);

    auto send_target = new SendAnswerTargetProtocol();

    bool check = false;
    if (publish_user && publish_user->m_is_publish_stream)
    {
        check = publish_user->m_publish_set.find(user_id) != publish_user->m_publish_set.end();
    }
    if (check && subscribe_user)
    {
        send_target->MakeResponseHeaderProtocol(send_answer_target_k, data->m_room->RoomId(), TARGET, 0);

        send_target->m_response_header->m_from_user_id = user_id;
        send_target->m_response_header->m_to_user_id   = target_user_id;

        send_target->m_sdp = sdp_info;
        send_target->m_type= sdp_type;
        send_target->m_user_id = user_id;
    }
    else
    {
        send_target->MakeResponseHeaderProtocol(send_answer_target_k, data->m_room->RoomId(), TARGET, -1);
    }
    data->m_response_vec.PushBack(send_target);
}

void PublishSubscribeModule::IceCandidateReq(RoomReferenceData *data, std::shared_ptr<BaseProtocol> &request)
{
    auto * req_protocol = (IceCandidateReqProtocol*)request.get();

    std::string & target_user_id = req_protocol->m_user_id;
    auto & candidate      = req_protocol->m_candidate;

    std::string & user_id        = data->m_connection_data->m_user_name;

    auto send_target = new IceCandidateTargetProtocol();

    send_target->MakeResponseHeaderProtocol(ice_candidate_target_k, data->m_room->RoomId(), TARGET, 0);
    send_target->m_response_header->m_from_user_id = user_id;
    send_target->m_response_header->m_to_user_id   = target_user_id;

    send_target->m_user_id   = user_id;
    send_target->m_candidate = candidate;
    send_target->m_sdp_type  = req_protocol->m_sdp_type;

    data->m_response_vec.PushBack(send_target);
}
